package com.zhss.dfs.upgrade;

import java.io.IOException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @description:
 * @Author liu Weidong
 * @Email:
 * @Date: 2024/5/31 9:40
 * @Version 1.0
 */
public class Processor extends Thread {

    public static final Long POLL_BLOCK_MAX_TIME = 1000L;

    /**
     * 初测连接的请求
     */
    private ConcurrentLinkedQueue<SocketChannel> queue = new ConcurrentLinkedQueue<>();

    /**
     * 缓存没有读完的请求
     */
    private Map<String, NetworkRequest> cacheRequest = new ConcurrentHashMap<>();

    /**
     * 缓存响应求情
     */
    private Map<String, NetworkResponse> cacheResponse = new ConcurrentHashMap<>();


    private Map<String, SelectionKey> cacheKey = new ConcurrentHashMap<>();

    private Selector selector;

    private String processorId;

    public Processor(String processorId) {
        try {
            this.selector = Selector.open();
            this.processorId = processorId;
        } catch (IOException ioException) {
            ioException.printStackTrace();
        }
    }

    public void addChannel(SocketChannel socketChannel) {
        queue.offer(socketChannel);
        selector.wakeup();
    }

    @Override
    public void run() {
        while (true) {
            try {
                // 进行排队等待注册地
                registerQueuedClients();
                // 缓存响应事件
                cacheQueueResponse();
                // 轮训方式解析请求中的时间
                poll();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private void cacheQueueResponse() {
        NetworkResponseQueue networkResponseQueue = NetworkResponseQueue.get();
        NetworkResponse networkResponse = null;
        if ((networkResponse = networkResponseQueue.poll(processorId)) != null) {
            cacheResponse.put(networkResponse.getClient(), networkResponse);
            cacheKey.get(networkResponse.getClient()).interestOps(SelectionKey.OP_WRITE);
        }
    }

    private void poll() {
        try {
            int keys = selector.select(POLL_BLOCK_MAX_TIME);
            if (keys > 0) {
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    SocketChannel socketChannel = (SocketChannel) key.channel();
                    String client = socketChannel.getRemoteAddress().toString();
                    // 接收到某个客户端的请求
                    if (key.isReadable()) {
                        // 有缓存则从缓存中取，没有新建
                        NetworkRequest request = cacheRequest.containsKey(client) ? cacheRequest.get(client) : new NetworkRequest();
                        request.setSocketChannel(socketChannel);
                        request.setKey(key);
                        request.read();
                        if (request.hasCompletedRead()) {
                            request.setClient(client);
                            request.setProcessorId(processorId);
                            NetworkRequestQueue.get().offer(request);
                            cacheKey.put(client,key);
                            // 请求构造完毕
                            cacheRequest.remove(client);
                            key.interestOps(key.interestOps() &~ SelectionKey.OP_READ);
                        } else {
                            cacheRequest.put(client, request);
                        }
                    } else if (key.isWritable()) {
                        NetworkResponse response = cacheResponse.get(client);
                        socketChannel.write(response.getBuffer());
                        cacheResponse.remove(client);
                        cacheKey.remove(client);
                        key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void registerQueuedClients() {
        SocketChannel channel = null;
        try {
            while ((channel = queue.poll()) != null) {
                channel.register(selector, SelectionKey.OP_READ);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
